package com.mbigger.redis.controller;

import com.mbigger.redis.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

@RestController
public class IndexController {

    private final static String KEY = "my-stream";
    private final static String SKEY = "s-stream";
    @Autowired
    RedisTemplate<String, String> redisTemplate;
    @Autowired
    RedisTemplate<String, User> userRedisTemplate;

    @GetMapping("/add/{content}")
    private String add(@PathVariable String content) {
        ObjectRecord<String, String> record = StreamRecords.newRecord().in(SKEY).ofObject(content);
        redisTemplate.opsForStream().add(record);
        return content;
    }

    @GetMapping("/read")
    private String read() {
        List<ObjectRecord<String, String>> msg = redisTemplate.opsForStream().read(String.class, StreamOffset.fromStart(SKEY));
        return msg.get(0).getValue();
    }

    @GetMapping("/range/{from}")
    private List<ObjectRecord<String, String>> range(@PathVariable String from) {
        RecordId oldId = RecordId.of(from);
        RecordId newId = RecordId.of(oldId.getTimestamp(), oldId.getSequence() + 1);
        List<ObjectRecord<String, String>> msg = redisTemplate.opsForStream().range(String.class, SKEY, Range.rightUnbounded(Range.Bound.inclusive(newId.getValue())));
        return msg;
    }

    @GetMapping("/xadd")
    private String xadd() {
        User user = new User();
        user.setFirt("f");
        user.setLast("l");
        ObjectRecord<String, User> record = StreamRecords.newRecord().in(KEY).ofObject(user);
        redisTemplate.opsForStream().add(record);
        return user.toString();
    }

    @GetMapping("/xread")
    private String xread() {
        List<ObjectRecord<String, User>> msg = redisTemplate.opsForStream().read(User.class, StreamOffset.fromStart(KEY));
        return msg.get(msg.size() - 1).toString();
    }
}
